package com.ikas.ai.server.kafka;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONValidator;
import com.ikas.ai.server.module.instance.model.DorisWorkFlowInstance;
import com.ikas.ai.server.module.instance.model.DorisWorkInstanceStep;
import com.ikas.ai.server.module.instance.service.DorisWorkFlowInstanceService;
import com.ikas.ai.server.module.instance.service.DorisWorkInstanceStepCostService;
import com.ikas.ai.utils.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

@Slf4j
@Component
public class WorkFlowInstanceStepAvgCostTimeConsumer {

    @Autowired
    DorisWorkFlowInstanceService workFlowInstanceService;
    @Autowired
    DorisWorkInstanceStepCostService workInstanceStepCostService;

    @KafkaListener(topics = KafkaInitialConfiguration.WORK_INSTANCE_STEP_DELAY, groupId = KafkaInitialConfiguration.WORK_INSTANCE_STEP_DELAY_GROUP, topicPattern = "0")
    private void processKafkaMessageForStep(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        Instant start = Instant.now();
        list.forEach(record -> {
            try {
                List<DorisWorkInstanceStep> stepList = new ArrayList<>();
                final JSONValidator.Type type = JSONValidator.from(record.value()).getType();
                if (type == JSONValidator.Type.Array) {
                    stepList = JSONArray.parseArray(record.value(), DorisWorkInstanceStep.class);
                } else if (type == JSONValidator.Type.Object) {
                    stepList = Collections.singletonList(JSONObject.parseObject(record.value(), DorisWorkInstanceStep.class));
                }
                if (CollectionUtil.isEmpty(stepList)) {
                    return;
                }
                log.info("====STEP-KAFKA-====>offset>[{}]--timestamp>[{}]------>meteDataList:{}", record.offset()
                        , record.timestamp(), JSONUtil.toJsonStr(stepList));
                stepList.forEach(this::generateStepAvgCostTime);
            } catch (Exception e) {
                log.error("=====Step平均耗时计算：消费错误======", e);
            }
        });
        //手动提交
        ack.acknowledge();
        Instant end = Instant.now();
        log.info("工况步骤时长kafka消费执行时长毫秒：{}", Duration.between(start, end).toMillis());
    }

    /**
     * 刷新Step平均耗时表
     *
     * @param step
     */
    private void generateStepAvgCostTime(DorisWorkInstanceStep step) {
        DorisWorkFlowInstance flowInstance = workFlowInstanceService.getById(step.getWfInstanceId());
        if (!ObjectUtil.isNull(flowInstance)) {
            workInstanceStepCostService.computeAvgDelay(flowInstance.getMachineNo(), step);
        }
    }

}
